package com.cbh.amadeus.client.receiver.consumer;

import com.cbh.amadeus.client.annotation.AmadeusMessageRegister;
import com.cbh.amadeus.client.domain.MessageMatchModel;
import com.cbh.amadeus.client.domain.MessageResultModel;
import com.cbh.amadeus.client.receiver.service.MessageHolder;
import com.cbh.amadeus.client.receiver.tool.MessageResultModelTool;
import com.cbh.amadeus.client.utils.ChannelReference;
import com.cbh.amadeus.client.utils.MessageHolderProxyChange;
import com.cbh.amadeus.common.enums.MessageResult;
import com.cbh.amadeus.common.json.Json;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.utils.StringUtil;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.Executor;

/**
 * @author ChenBiHui
 * @date 2025/4/28
 */
@Slf4j
public class MessageHolderWrap {

    private static final Method METHOD;

    static {
        try {
            METHOD = MessageHolder.class.getMethod("attain", String.class, String.class, String.class, Object.class);
            METHOD.setAccessible(true);
        } catch (NoSuchMethodException e) {
            throw new RuntimeException(e);
        }
    }

    private final ApplicationContext context;
    private final MessageResultModelTool tool;

    public MessageHolderWrap(ApplicationContext context, MessageResultModelTool tool) {
        this.context = context;
        this.tool = tool;
    }

    public void match(Channel channel, String name, AmadeusMessage message, List<MessageHolder<?>> holders) {
        Object payload = holders.stream()
                .findAny()
                .map(item -> Json.read(MessageHolder.class, item.getClass(), message.getPayload()))
                .orElse(null);
        if (payload == null) {
            log.warn(
                    "===== Topic [{}], this message can't read json to object, message's content is [{}] =====",
                    message.getTopic(),
                    message.getPayload()
            );
            return;
        }
        ChannelReference.positioner(channel, name, message, holders.size());
        for (MessageHolder<?> holder : holders) {
            AmadeusMessageRegister register = AnnotationUtils.findAnnotation(holder.getClass(), AmadeusMessageRegister.class);
            if (register == null) {
                log.warn(
                        "===== Current [MessageHolder] [{}], can't find @MessageRegister annotation =====",
                        holder.getClass()
                );
                continue;
            }
            if (StringUtil.unHasText(register.executor())) {
                callback(name, message, payload, holder);
            } else {
                Executor executor = context.getBean(register.executor(), Executor.class);
                executor.execute(() -> callback(name, message, payload, holder));
            }
        }
    }

    private void callback(String name, AmadeusMessage message, Object payload, MessageHolder<?> holder) {
        MessageResult notified = notify(holder, message, payload);
        try {
            MessageResultModel resultModel = new MessageResultModel();
            resultModel.setId(tool.sequence());
            resultModel.setTenant(message.getTenant());
            resultModel.setTopic(message.getTopic());
            resultModel.setMessageId(message.getMessageId());
            resultModel.setConsumer(name);
            resultModel.setLabel(MessageHolderProxyChange.get(holder));
            resultModel.setInvokeResult(notified.name());
            resultModel.setInvokeTime(LocalDateTime.now());
            tool.retain(resultModel);
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            return;
        }
        MessageMatchModel seek = ChannelReference.seek(message);
        if (seek == null) {
            return;
        }
        seek.mount(notified, holder.getClass().getName());
    }

    private MessageResult notify(MessageHolder<?> holder, AmadeusMessage message, Object payload) {
        try {
            Object result = METHOD.invoke(holder, message.getTenant(), message.getTopic(), message.getMessageId(), payload);
            if (result == null) {
                return MessageResult.UNKNOWN;
            }
            return (MessageResult) result;
        } catch (InvocationTargetException target) {
            Throwable throwable = target.getTargetException();
            log.error(throwable.getMessage(), throwable);
            return MessageResult.FAILURE;
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            return MessageResult.FAILURE;
        }
    }
}
